package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 用户行为日志数据： --app 启动日志--
 * 	【需求：启动日志中，各个广告播放总时长】
 * 	    {
 * 	  "common": {
 * 	    "ar": "310000",
 * 	    "ba": "Xiaomi",
 * 	    "ch": "vivo",
 * 	    "is_new": "1",
 * 	    "md": "Xiaomi 9",
 * 	    "mid": "mid_816719",
 * 	    "os": "Android 10.0",
 * 	    "uid": "174",
 * 	    "vc": "v2.1.134"
 *      },
 * 	  "start": {
 * 	    "entry": "icon",
 * 	    "loading_time": 14616,
 * 	    "open_ad_id": 15,
 * 	    "open_ad_ms": 6297,
 * 	    "open_ad_skip_ms": 0
 *      },
 * 	  "ts": 1713404820000
 *    }
 * @author xuanyu
 * @date 2025/10/21
 */
public class _05RealtimeStartDemo {

	public static void main(String[] args) throws Exception {
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("topic-log")
			.setGroupId("my-group-1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);
//		stream.print("kafka");


		// 3. 数据转换-transformation
		// 3-1. 获取启动日志数据
		SingleOutputStreamOperator<String> stream31 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// start字段值
					Object startValue = jsonObject.get("start");
					if(null != startValue){
						// 解析
						JSONObject startJsonObject = jsonObject.getJSONObject("start");
						// 提取字段值
						Object openAdIdValue = startJsonObject.get("open_ad_id");
						if(null != openAdIdValue){
							return true ;
						}
					}
					return false;
				}
			}
		);

		// 3-2. 提取字段值，封装二元组
		SingleOutputStreamOperator<Tuple2<Integer, Long>> stream32 = stream31.map(
			new MapFunction<String, Tuple2<Integer, Long>>() {
				@Override
				public Tuple2<Integer, Long> map(String value) throws Exception {
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 解析
					JSONObject startJsonObject = jsonObject.getJSONObject("start");
					// 字段值
					Integer openAdIdValue = startJsonObject.getInteger("open_ad_id");
					Long openAdMsValue = startJsonObject.getLong("open_ad_ms");
					// 返回
					return Tuple2.of(openAdIdValue, openAdMsValue);
				}
			}
		);

		// 3-3. 分组
		KeyedStream<Tuple2<Integer, Long>, Integer> stream33 = stream32.keyBy(
			new KeySelector<Tuple2<Integer, Long>, Integer>() {
				@Override
				public Integer getKey(Tuple2<Integer, Long> value) throws Exception {
					return value.f0;
				}
			}
		);

		// 3-4. 聚合
		SingleOutputStreamOperator<Tuple2<Integer, Long>> stream34 = stream33.sum(1);
		//stream34.print();

		// 4.数据接收器-sink
/*
USE gmall_report;
CREATE TABLE IF NOT EXISTS start_ad_report(
    open_ad_id int PRIMARY KEY ,
    open_ad_ms long
);

REPLACE INTO gmall_report.start_ad_report(open_ad_id, open_ad_ms) VALUES (?, ?) ;


 */
		SinkFunction<Tuple2<Integer, Long>> jdbcSink = JdbcSink.sink(
			"REPLACE INTO gmall_report.start_ad_report(open_ad_id, open_ad_ms) VALUES (?, ?)",
			(statement, tuple) -> {
				statement.setObject(1, tuple.f0);
				statement.setObject(2, tuple.f1);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/gmall_report")
				.withDriverName("com.mysql.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream34.addSink(jdbcSink) ;

		// 5. 触发执行
		env.execute("RealtimeStartDemo") ;
	}

}
